"""
读取文件转换成rdd，并完成
打印输出：热门搜索时间段（小时精度）top3
打印输出：热门搜索关键词top3
打印输出：统计黑马程序员关键字在那个时间段（小时精度）搜索次数最多
将数据转换为json格式保存到文件中
"""
from pyspark import SparkConf, SparkContext
import os
import json
os.environ['PYSPARK_PYTHON'] = "D:/dev/python/python310/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("综合案例")
sc = SparkContext(conf=conf)
# 读取数据
rdd = sc.textFile("search_log.txt")
# 数据清洗
rdd2 = rdd.map(lambda x: x.split("\t"))
# 热门搜索时间段（小时精度 ）top3
rdd3 = rdd2.map(lambda x: (x[0][:2], 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .sortBy(lambda x: x[1], ascending=False)
print("热门搜索时间段（小时精度）top3:", rdd3.take(3))
# 热门搜索关键词top3
rdd4 = rdd2.map(lambda x: (x[2], 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .sortBy(lambda x: x[1], ascending=False)
print("热门搜索关键词top3:", rdd4.take(3))
# 统计黑马程序员关键字在那个时间段（小时精度）搜索次数最多
rdd5 = rdd2.filter(lambda x: "黑马程序员" in x[2]) \
    .map(lambda x: (x[0][:2], 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .sortBy(lambda x: x[1], ascending=False)
print("黑马程序员关键字在那个时间段（小时精度）搜索次数最多:", rdd5.take(1))
# 将数据转换为json格式保存到文件中
result = {
    "热门搜索时间段top3": rdd3.take(3),
    "热门搜索关键词top3": rdd4.take(3),
    "黑马程序员搜索时间段": rdd5.take(1)
}
with open("search_log_result.json", "w", encoding="utf-8") as f:
    json.dump(result, f, ensure_ascii=False, indent=4)